package cc.verywell.pureblock.msg.nfs;

import java.nio.channels.CompletionHandler;
import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import cc.verywell.pureblock.func.node.KeyShell;
import cc.verywell.pureblock.func.node.NodeBucket;
import cc.verywell.pureblock.func.node.NodeBucket.ValueShell;
import cc.verywell.pureblock.log.Pblog;
import cc.verywell.pureblock.net.AioClientImpl0;
import cc.verywell.pureblock.net.AioClientConnector;
import cc.verywell.pureblock.net.AioClientInterface;
import cc.verywell.pureblock.net.NormalSession;

/**
* 汇总更新程序;
* 
* Copyright © 2020 TanYaqiu. All rights reserved.
* @author TanYaqiu 
* @address flat 601
* @date 2020年2月6日 下午11:04:14
*/
public class NfsPolicyV1newStoreHandler implements CompletionHandler<NormalSession, NfsPolicyV1newStoreHandler.RemoteNode>{
	static public long REMOTE_NEWORDERCHECK_TIMEOUT = 30000;
	public static boolean DEBUG = true;
	
	private NfsFuncProvider nfs;
	private AioClientInterface<NormalSession, AioClientConnector> client;
	public void startConnect(NfsFuncProvider nfs, AioClientInterface<NormalSession, AioClientConnector> client) { //开始执行:
		this.nfs = nfs;
		this.client = client;
		finishAdd();
		if(Pblog.DEBUG&&DEBUG) Pblog.DEBUG(this.getClass().getSimpleName(),"(",Integer.toHexString(hashCode()),")::startConnect( ",fileHashs.size(),", nodes:",getNotNullCount(remoteNodes),"/",remoteNodes.length);
		startNewConnect();
	}
	
	@Override
	public void completed(NormalSession ns, RemoteNode r) {
		if(Pblog.DEBUG&&DEBUG)Pblog.DEBUG(this.getClass().getSimpleName(),"(",Integer.toHexString(hashCode()),")::completed( ",fileHashs.size(),", nodes:",getNotNullCount(remoteNodes),"/",remoteNodes.length);
		boolean ok = false;
		try {
			ok = writeMsg(ns, r);
		}finally {
			if(ok)remoteCompleted(r);
			else remoteFailed(r);
		}
		if(Pblog.DEBUG&&DEBUG)Pblog.DEBUG(this.getClass().getSimpleName(),"(",Integer.toHexString(hashCode()),")::completed( ",fileHashs.size(),", nodes:",getNotNullCount(remoteNodes),"/",remoteNodes.length);
		startNewConnect();
	}
	@Override
	public void failed(Throwable exc, RemoteNode r) {
		if(Pblog.DEBUG&&DEBUG)Pblog.DEBUG(this.getClass().getSimpleName(),"(",Integer.toHexString(hashCode()),")::failed( ",fileHashs.size(),", nodes:",getNotNullCount(remoteNodes),"/",remoteNodes.length);
		remoteFailed(r);
		if(Pblog.DEBUG&&DEBUG)Pblog.DEBUG(this.getClass().getSimpleName(),"(",Integer.toHexString(hashCode()),")::failed( ",fileHashs.size(),", nodes:",getNotNullCount(remoteNodes),"/",remoteNodes.length);
		startNewConnect();
	}

	
	protected void startNewConnect() {
		RemoteNode r;
		while(null!=(r = findmostRelative())) {
			ValueShell v = r.remoteNode;
			NormalSession ns = v.getSession();
			if(null!=ns && !ns.isClosed()) {
				boolean ok = false;
				try {
					ok = writeMsg(ns, r);
				}finally {
					if(ok)remoteCompleted(r);
					else remoteFailed(r);
				}
				continue;
			}
			String[] remote = v.getRemote();
			if(!ValueShell.isConnectable(remote)) {
				remoteFailed(r);
				continue;
			}
			AioClientInterface<NormalSession, AioClientConnector> client = this.client;
			if(client==null)return;
			client.connect(v, r, this);
			return;
		}
	}
	private synchronized RemoteNode findmostRelative() {
		RemoteNode mostNode = null;
		int mostRelated = -1;
		for (RemoteNode r : remoteNodes) {
			if(r!=null && r.related > mostRelated) {
				mostNode = r;
				mostRelated = r.related;
			}
		}
		return mostNode;
	}
	private synchronized void remoteCompleted(RemoteNode r) {
		List<FileHash> okFileHashs = new LinkedList<FileHash>();
		if(remoteNodes[r.idx]!=null) {
			remoteNodes[r.idx]=null;
			int off = r.idx>>>5;
			int mth = 1<<(r.idx&MASK32);
			int rest = r.related;
			Iterator<FileHash> it = fileHashs.iterator();
			while(it.hasNext()) {
				FileHash f = it.next();
				if(f.matrix.length>off && 0!=(f.matrix[off]&mth)) {
					f.related--;
					f.okrest--;
					if(f.related==0)
						it.remove();
					else if(f.okrest==0) {
						it.remove();
						okFileHashs.add(f);
					}
					if(--rest==0)break;
				}
			}
		}
		for (FileHash f : okFileHashs) {
			dropedFile(f);
		}
	}
	private synchronized void dropedFile(FileHash f) { //在fileHashs里remove以后再运行;
		//已通过LangTest#testMatric测试
		int[] matrix = f.matrix;
		int rest = f.related;
		f:for (int off = 0; off < matrix.length; off++) {
			int src = matrix[off];
			if(matrix[off]==0)continue;
			for (int mth=0,cmp=1; mth<=MASK32; mth++,cmp<<=1) {
				if(0!=(src&cmp)) {
					int idx = (off<<5)|mth;
					if(remoteNodes[idx]!=null) {
						if( 0 == --remoteNodes[idx].related) {
							remoteNodes[idx]=null;
						}
						if(--rest==0)break f;
					}
				}
			}
		}
	}
	private synchronized void remoteFailed(RemoteNode r) {
		if(remoteNodes[r.idx]!=null) {
			remoteNodes[r.idx]=null;
			int off = r.idx>>>5;
			int mth = 1<<(r.idx&MASK32);
			int rest = r.related;
			Iterator<FileHash> it = fileHashs.iterator();
			while(it.hasNext()) {
				FileHash f = it.next();
				if(f.matrix.length>off && 0!=(f.matrix[off]&mth)) {
					f.related--;
					if(f.related==0)it.remove();
					if(--rest==0)break;
				}
			}
		}
	}
	
	
	
	private AtomicInteger tempNewidx;
	private Map<KeyShell,RemoteNode> tempNodesMap;
	private List<FileHash> fileHashs = new LinkedList<FileHash>();
	private RemoteNode[] remoteNodes; //在finishAdd后添加;
	public NfsPolicyV1newStoreHandler() {
		tempNewidx = new AtomicInteger(0);
		tempNodesMap = new ConcurrentHashMap<>();
	}
	private static final int MASK32 = 0B11111;/*==5*/
	public void add(KeyShell fileHash,List<NodeBucket.ValueShell> nodes,int okrest) { //可在并行中写入 , 此时fileHash已合并;
		int remoteIdxLength = 0;
		int[] remoteIdx = new int[nodes.size()];
		int maxidx = -1;
		for (NodeBucket.ValueShell n : nodes) {
			RemoteNode r = tempNodesMap.computeIfAbsent(n.key, k->{
				return new RemoteNode(tempNewidx.getAndIncrement());
			});
			remoteIdx[remoteIdxLength++]=r.idx;
			synchronized (r) {
				maxidx = maxidx>r.idx?maxidx:r.idx;
				r.remoteNode = n;
				r.related++;
			}
		}
		FileHash fh = new FileHash();
		fh.filehash = fileHash;
		fh.okrest = okrest;
		fh.related = remoteIdx.length;
		//已通过LangTest#testMatric
		int ml = (maxidx>>>5)+1;
		int[] matrix = new int[ml];
		for (int i : remoteIdx) {
			matrix[i>>>5] |= (1<<(i&MASK32));
		}
		fh.matrix = matrix;
		
		synchronized (fileHashs) {
			fileHashs.add(fh);
		}
		if(Pblog.DEBUG&&DEBUG)Pblog.DEBUG(this.getClass().getSimpleName(),"(",Integer.toHexString(hashCode()),")::add( ",Integer.toHexString(fileHash.hashCode()),"/",fileHashs.size(),", nodes:",(nodes==null?"null":nodes.size()),"/",tempNodesMap.size(),", okrest:",okrest,")");
	}
	private synchronized void finishAdd() {
		remoteNodes = new RemoteNode[tempNewidx.get()];
		for (RemoteNode r : tempNodesMap.values()) {
			remoteNodes[r.idx] = r;
		}
		tempNewidx = null;
		tempNodesMap = null;
	}
	
	
	public static class FileHash{
		KeyShell filehash;
		int okrest;
		long lastCheckTime;
		int[] matrix;
		int related;
	}
	public static class RemoteNode implements Comparable<RemoteNode>{
		final int idx;
		ValueShell remoteNode;
		int related = 0;
		public RemoteNode(int idx) {
			this.idx = idx;
		}
		@Override
		public int compareTo(RemoteNode o) {
			return this.idx - o.idx;
		}
	}
	
	
	
	
	private boolean writeMsg(NormalSession ns, RemoteNode r) {
		NfsFuncProvider nfs = this.nfs;
		if(nfs==null || nfs.fsmap==null)return false;
		
		List<FileHash> okFileHashs = new LinkedList<FileHash>();
		synchronized (this) {
			int off = r.idx>>>5;
			int mth = 1<<(r.idx&MASK32);
			int rest = r.related;
			Iterator<FileHash> it = fileHashs.iterator();
			while(it.hasNext()) {
				FileHash f = it.next();
				if(f.matrix.length>off && 0!=(f.matrix[off]&mth)) {
					okFileHashs.add(f);
					if(--rest==0)break;
				}
			}
		}
		
		
		List<byte[]> ls;
		if(okFileHashs.size()>1) { //并行处理; 当文件数量大于1时
			List<FileHash> lessFileHashs = new LinkedList<FileHash>();
			try(
				Stream<FileHash> stream = okFileHashs.parallelStream();
			){
				ls = stream.filter(f->{ //并行检查文件;
					boolean check = false;
					synchronized (f) {
						if(System.currentTimeMillis()-f.lastCheckTime>REMOTE_NEWORDERCHECK_TIMEOUT) { //每30秒检查一次看看文件还在不在
							check = true;
							f.lastCheckTime = System.currentTimeMillis();
						}
					}
					if(check && !nfs.fsmap.contain(f.filehash)) {
						synchronized (this) {
							if(fileHashs.remove(f)) {
								dropedFile(f);
							}
						}
						return false;
					}else {
						return true;
					}
				}).map(f->f.filehash.key).collect(Collectors.toList());
				if(ls.isEmpty()) {
					return false;
				}
			}
		}else if(okFileHashs.size()>0) { //当文件数量等于1时
			FileHash f = okFileHashs.get(0);
			boolean check = false;
			synchronized (f) {
				if(System.currentTimeMillis()-f.lastCheckTime>REMOTE_NEWORDERCHECK_TIMEOUT) { //每30秒检查一次看看文件还在不在
					check = true;
					f.lastCheckTime = System.currentTimeMillis();
				}
			}
			if(check && !nfs.fsmap.contain(f.filehash)) {
				synchronized (this) {
					if(fileHashs.remove(f)) {
						dropedFile(f);
					}
				}
				return false;
			}
			ls = Collections.singletonList(f.filehash.key);
		}else {
			ls = null;
			return false;
		}

		if(ns!=null && ns.getFunc().hasFunc(nfs)) {
			NfsFuncProvider.StoreFileMsg m = nfs.new StoreFileMsg(ls);
			if(DEBUG&&Pblog.DEBUG)Pblog.DEBUG("send newStore:",ls.size()," at ",ns," by ",this.getClass().getSimpleName(),"(",Integer.toHexString(hashCode()),")");
			ns.writeMsg(m);
			return true;
		}else {
			return false;
		}
	}
	
	static private int getNotNullCount(Object[] os) {
		int i = 0;for(Object o : os)if(o!=null)i++;return i;
	}

}
